package org.jetlinks.pro.things.impl;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.MetadataId;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.core.things.ThingRpcSupport;
import org.jetlinks.pro.things.ThingConstants;
import org.jetlinks.pro.things.ThingsConnector;
import org.jetlinks.pro.things.impl.connector.AbstractThingsConnector;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.function.Function;

/**
 * 递归检查物连接器,用于检查连接器是否存在递归,当出现递归时,不再输出结果
 *
 * @author zhouhao
 * @see ThingConstants#connectorTrace
 * @since 1.12
 */
@AllArgsConstructor(staticName = "of")
@Slf4j
public class RecursiveCheckThingsConnector extends AbstractThingsConnector
    implements ThingsConnector, ThingRpcSupport {

    //递归检查连接器的跟踪标识
    private final String traceId;

    //需要递归检查的连接器
    private final ThingsConnector source;

    /**
     * 通过物消息header中的跟踪标识判断物消息是否存在递归
     *
     * @param message 连接器中的物消息
     * @return 是否存在递归的结果
     */
    private boolean hasRecursion(ThingMessage message) {
        if (message
            .getOrAddHeaderDefault(ThingConstants.connectorTrace)
            .add(traceId)) {
            return false;
        }
        log.debug("thing message has recursion {}", message);
        return true;
    }

    /**
     * 连接到指定物的物模型,当产生消息时,handler将被调用.可调用返回值的{@link Disposable#dispose()}来断开连接
     *
     * @param thingId    物ID
     * @param metadataId 物模型ID
     * @param handler    消息处理器
     * @return Disposable
     */
    @Override
    protected Disposable doConnectTo(ThingId thingId,
                                     MetadataId metadataId,
                                     Function<ThingMessage, Mono<Void>> handler) {
        return source
            .connectTo(thingId, metadataId, msg -> {
                //如果存在递归则不在做处理，返回空流
                if (hasRecursion(msg)) {
                    return Mono.empty();
                }
                return handler.apply(msg);
            });
    }

    /**
     * 连接到当前物的物模型.物实例ID由数据源生成. 通常用于批量连接处理,比如在物模版里配置规则,计算对应物实例
     * 当产生消息时,handler将被调用.可调用返回值的{@link Disposable#dispose()}来断开连接
     *
     * @param metadataId 物模型ID
     * @param handler    消息处理器
     * @return Disposable
     */
    @Override
    protected Disposable doConnectTo(MetadataId metadataId,
                                     Function<ThingMessage, Mono<Void>> handler) {
        return source
            .connectTo(metadataId, msg -> {
                //如果存在递归则不在做处理，返回空流
                if (hasRecursion(msg)) {
                    return Mono.empty();
                }
                return handler.apply(msg);
            });
    }

    /**
     * 释放连接器资源
     */
    @Override
    protected void onDispose() {
        source.dispose();
    }

    /**
     * 判断是否包装自指定的类型,通常在将当前接口转为特定的类型时进行判断.
     *
     * @param target 目标类型
     * @return true or false
     */
    @Override
    public boolean isWrapperFor(Class<?> target) {
        if (target == RecursiveCheckThingsConnector.class) {
            return true;
        }
        if (target == ThingRpcSupport.class) {
            return true;
        }
        return source.isWrapperFor(target);
    }

    /**
     * 将当前接口拆包为指定的类型
     *
     * @param target 目标类型
     * @return 指定的类型
     * @throws ClassCastException 当无法转为指定类型时抛出
     */
    @Override
    public <T> T unwrap(Class<T> target) throws ClassCastException {
        if (target == RecursiveCheckThingsConnector.class) {
            return target.cast(this);
        }
        if (target == ThingRpcSupport.class) {
            return target.cast(this);
        }
        return source.unwrap(target);
    }

    /**
     * 将请求消息转换为对应物的连接器支持
     *
     * @param message 请求消息
     * @return 物连接器
     */
    @Override
    public Flux<? extends ThingMessage> call(ThingMessage msg) {
        //如果存在递归则不在做处理，返回空流
        if (hasRecursion(msg)) {
            return Flux
                .empty();
        }
        return source.unwrap(ThingRpcSupport.class).call(msg);
    }
}
